#!/usr/bin/env node
/* 
 * ProjectFile: message-broker\src\msg.server.js
 * Created: lijianwei 2021-08-03 22:51:17
 * -----
 * LastModified: lijianwei 2023-01-10 10:47:44
 * -----
 * Copyright (c) 2016-2023 Rongwen Tech Ltd.
 * -----
 */
import { createServer } from 'websocket-stream'
import http from 'http'
import aedes from 'aedes'
import utils from 'aedes/lib/utils.js'
import net from 'net'
import { getUserInfo } from './api/index.js'
import { formatDate, toMerge } from './utils/tools.js'
import { ErrorResult } from './utils/ErrorResult.js'
import CONFIG from './config.cjs'
const { AppConfig } = CONFIG // cjs的模块不支持直接析构

console.log("AppConfig", AppConfig) //csl

/**
 * 在线推送broker 
 */
const mqttPort = parseInt(AppConfig.mqttPort)
    , wsPort = parseInt(AppConfig.wsPort)
    , listenHost = '0.0.0.0'

// http服务
const httpServer = http.createServer((request, response) => {

})

const mqttServ = aedes({
    // mq 
    // concurrency 
    // persistence 
    // queueLimit 
    // maxClientsIdLength 
    // id - broker
    heartbeatInterval: 60 * 1000, //心跳检测间隔
    connectTimeout: 20 * 1000, //connect_ack等待时间
})
/**
 *  api 
 *  connectedClients 在连接中的客户端
 *  brokers          [分布式]使用中服务器
 */

// mqtt服务
const server = net.createServer(mqttServ.handle)
// mqtt监听
server.listen(mqttPort, listenHost, () => { //mqtt服务启动后
    console.log('[PushServer-mqtt] port-listened:', mqttPort)
})


// 升级httpServer 为 websocket协议; 并用mqtt协议 创建 mqtt over websocket
createServer({
    server: httpServer
}, mqttServ.handle)

// mqtt over websocket 监听
httpServer.listen(wsPort, listenHost, () => { // ws服务启动后
    console.log('[PushServer-ws] port-listened:', wsPort)
})

// 连接断开 - 心跳超时
mqttServ.on('clientError', (client, err) => {
    AppConfig.ENV_DEBUG && console.log(formatDate(new Date().getTime()) + ' [clientError - keepaliveTimeout]', client.id, err.message, err.stack)
})

// 连接断开 - 无心跳超时
mqttServ.on('connectionError', (client, err) => {
    AppConfig.ENV_DEBUG && console.log(formatDate(new Date().getTime()) + ' [connectionError]', client.id, err.message, err.stack)
})

// 客户端连接-断链后
mqttServ.on('clientDisconnect', (client) => {
    AppConfig.ENV_DEBUG && console.log(formatDate(new Date().getTime()) + ' [clientDisconnect]', client.id)
})

// 客户端 authenticated
mqttServ.on('client', (client) => {
    AppConfig.ENV_DEBUG && console.log('[Connected - client.id]', client.id)
})

// version 1.0.1
/* Topic ACL 规则
AB 大类一同打包订阅
C 类在按客户端需求增/减订
D 读取用户的屏蔽设置来过滤主题, 默认订阅全部相关主题[ABC]

    [Persist 的是持久型, 订中间件队列服务]
    [Unpersist 的是非持久型, 订非持久二手处理服务]
    A 固定订阅
        ClientMsg/User/具体用户id

    B 用户体系 - 查具体数据决定
        ClientMsg/Usergrp/具体用户组id
        ClientMsg/Depart/具体部门id
        ClientMsg/Organization/具体企业id
        ClientMsg/Staff/具体工作人员id

    C 活动订阅
        ClientMsg/custom/活动唯一标识
        ClientMsg/custom/taskxxxx 有条件 与 期限
 */

//订阅的业务大类
// const subscribableTopic_cat = [
//     'ClientMsg', // 客户端业务信息
//     'WillMsg', // 遗愿讯息
// ]

// 主题前缀ACL认证规则 idProfix + idPreg主题字符校验 + validateCall 标识要通过认证中心校验其值的存在性
const subscribableTopic_rule = {
    'Client': {
        name: '客户端',
        descr: '客户端',
        idProfix: [],
        idPreg: '\\w{6,}',
        validateCall: '',
    },
    'User': {
        name: '帐号',
        descr: '帐号',
        idProfix: [],
        idPreg: '\\d{1,}',
        validateCall: 'mapUserId',
    },
    'Custom': { // 其他一律用广播
        name: '场景广播',
        descr: '场景广播',
        idProfix: ['role', 'device', 'usergrp', 'event', 'conn'],
        idPreg: '',
        validateCall: '',
    },
}

/**
 * 主题合法检查 {主题acl}
 * @param {string} topic
 * @returns busResult
 * @version 1.0
 */
function topicCheck(topic, client) {
    // 没经过 Event:client 处理的 client 不会有 connInfo 字段
    if (!topic || !client.connInfo || !client.connInfo.id) {
        return new ErrorResult('数据有误!', 111)
    }

    const res = topicParse(topic)

    if (!res[1] || !subscribableTopic_rule[res[1]]) {
        console.log(res)
        return new ErrorResult('主题不合法!', 111)
    }

    // acl 

    return {
        msgCat: res[0], // 讯息技术大类
        objectType: res[1], // 场景分类
        objectId: res[2], // 场景细分标识 = idProfix / idPreg
    }
}


/**
 * 解析主题
 * @param {string} topic 
 * @returns 
 */
function topicParse(topic) {
    // 仅支持 Xxx/Yyy 主题
    return topic.split('/')
}

/**
 * 讯息发布
 * @param {*} currentMsg 
 * @param {*} topicOption 
 */
function distribute(currentMsg, topicOption, client, server) {
    const msg = toMerge({
        'code': 1, //1-正常, 其他-异常
        'body': currentMsg.body || {},
        'msg': currentMsg.msg || 'ok',
        'cmd': {
            'type': 1,//(客户端)指令类型, 1-基础连接管理, 默认2-自定义
            'action': 'consoleLog',
        },
        'time': formatDate(new Date().getTime()), //发布时间
    }, currentMsg)

    server.publish({
        cmd: 'publish',
        qos: typeof topicOption.qos === undefined ? 0 : topicOption.qos,
        topic: topicOption.topic,
        payload: Buffer.from(JSON.stringify(msg)),
        retain: typeof topicOption.retain === undefined ? false : topicOption.retain,
    })
}

/**
 * 错误清楚
 * @param {*} client 
 */
function cmdErrReset(client) {
    client.busError = {}
    client.connInfo = {}
}

//发布后, 含心跳
mqttServ.on('publish', (packet, client) => {
    if (packet.topic.startsWith(utils.$SYS_PREFIX)) { //系统原语: 心跳包 订阅包 取消订阅包 连接确认包, 断开确认包 ... 等等
        // AppConfig.ENV_DEBUG && console.log(' [Published - sys - topic]', packet.topic)
        // AppConfig.ENV_DEBUG && console.log(' [Published - sys - payload]', packet.payload.toString())
    } else { //common message
        console.log(' [Published - topic]', packet.topic,)
        console.log(' [Message_payload]', packet.payload.toString())
    }
})

//(协议环节连接后) + 订阅鉴权通过后, 大部分交互反馈的位置
mqttServ.on('subscribe', (subscriptions, client) => {
    AppConfig.ENV_DEBUG && console.log(formatDate(new Date().getTime()) + ' [on-subscribe Subscriptions]', subscriptions)
    AppConfig.ENV_DEBUG && console.log('[ client.id ]', client.id)
    //response the client
    if (client.busError instanceof ErrorResult) {

        const targetTopic = 'ClientMsg/Client/' + client.id
        distribute({
            msg: '[ client.id:' + client.id + ' ]' + (client.busError.msg || '[ unkown - error ]'),
            cmd: {
                'type': 1,
                'action': client.busError.cmdAction || 'consoleLog',
            },
            body: client.busError.body || {}
        }, { topic: targetTopic }, client, mqttServ)
        cmdErrReset(client) //每次用完要重置
    } else if (AppConfig.ENV_DEBUG && !!client.connInfo && client.connInfo.clientCat < 200) {
        const targetTopic = subscriptions[0].topic
        distribute({
            'body': {
                subscriptions,
                'client.id': client.id
            },
            'msg': '[ msg-Topic:' + targetTopic + ' ] Subsc suc!🤝' + (!client.connInfo ? '' : client.connInfo.real_name),
            'cmd': {
                'type': 1, //(客户端)指令类型, 1-基础连接管理, 2-按协定注册的方法执行, 默认3-自定义
                'action': 'consoleLog',
            },
        }, { topic: targetTopic }, client, mqttServ)
    }
})

/**
 * 连接预处理
 * Invoked when server receives a valid CONNECT packet.
 *  
 * @param {*} client 
 * @param {*} packet CONNECT-packet
 * @param {*} callback 
 */
mqttServ.preConnect = (client, packet, done) => {
    // dev...
    // 1 Rate Limit / Throttle by client.conn.remoteAddress

    // 2 Check aedes.connectedClient to limit maximum connections

    // 3 IP blacklisting
    cmdErrReset(client)
    done(null, true)
}

/**
 * [连接预处理后] 连接认证
 * Invoked after preConnect.
 * Server parses the CONNECT packet, initializes client object which set client.id to match the one in CONNECT packet and extract username and password as parameters for user-defined authentication flow.
 * If invoked callback with no errors and successful be true, server authenticates client and continues to setup the client session.
 * If authenticated, server acknowledges a CONNACK with returnCode=0, otherwise returnCode=5. Users could define the value between 2 and 5 by defining a returnCode property in error object.
 * @param {*} client 
 * @param {*} username 
 * @param {*} password 
 * @param {*} done 
 * @returns void
 */
mqttServ.authenticate = async (client, username, password, done) => {

    if (!username) {
        const error = new Error("username is null!")
        error.returnCode = 4
        done(error, null)
        return
    }
    //client.req.headers
    // console.log('[ --- client --- ]', client.req.headers)
    // console.log('[ --- user --- ]', username)
    cmdErrReset(client) // 初始化属性

    if (username.startsWith('rwmsp-')) { // rwmsp-内部系统访问
        // 认证dev???
        client.connInfo = {
            clientCat: 201, // 客户端分类, 101-用户设备, 102-机器设备, 201-中台程序
            id: 1,
            real_name: username,
            tokenValidateTime: Date.now(),
        }
        done(null, true)
        return
    } else { // 客户端访问

        let pwd
        if (typeof password === 'string') {
            pwd = password
        } else {
            pwd = password.toString()
        }
        // 获取token认证结果
        try {
            const res = await getUserInfo(undefined, { headers: { Authorization: "Bearer " + pwd } })
            if (!res.code) { // 认证服务异常
                // error.returnCode = 4
                client.busError = new ErrorResult('[Authenticate - Service Error]')
                // client.busError.cmdAction = 'clientReconnectTryTimes' // 告知客户端 重试N次后不做重连接
                client.busError.cmdAction = 'consoleLog'
                client.busError.errCat = 'error'
                client.busError.body = {
                    dataError: res
                }
            } else if (res.code !== '00000') { // 认证校验不通过
                // error.returnCode = 3
                client.busError = new ErrorResult('[Authenticate Failure] 用户:' + username + ',token无效 - ' + pwd)
                client.busError.cmdAction = 'clientReconnect' // 告知客户端token无效, 需要刷新token并重连
                client.busError.errCat = 'debug'
                client.busError.body = {
                    dataError: res
                }
            } else if (username != res.data.id) { // 认证数据有误
                // error.returnCode = 4
                client.busError = new ErrorResult('[Authenticate Failure] 用户:' + username + ',token过期 - ' + pwd)
                client.busError.cmdAction = 'clientReconnect' // 客户端连接过期, 用户信息丢失
                client.busError.errCat = 'debug'
                client.busError.body = {
                    dataError: res
                }
            }
            // 连续失败的处理 dev...

            // if (mqttServ.clients[client.id]) {
            //     console.log('[Old Socket Connection is exsit]', client.id)
            //     mqttServ.clients[client.id].conn.destroy(); //需要告知[断开连接] 代码无效...
            // }

            // 同类客户端链接 实时剔除
            // dev...

            if (client.busError instanceof ErrorResult) {
                if (!client.busError.errCat) {
                    client.busError.errCat = 'log'
                }
                const _time = formatDate(new Date().getTime())
                // 错误信息输出
                if (client.busError.errCat === 'debug') {
                    if (AppConfig.DebugMode) {
                        console.log(_time, 'busError.message', client.busError.message)
                        console.log(_time, 'busError.body', client.busError.body)
                    }
                } else if (client.busError.errCat === "log") {
                    console.log(_time, 'busError.message', client.busError.message)
                    console.log(_time, 'busError.body', client.busError.body)
                } else { // error
                    console.error(_time, 'busError.message', client.busError.message)
                    console.error(_time, 'busError.body', client.busError.body)
                }

            } else {
                // 注入用户信息
                client.connInfo = {
                    clientCat: 101,
                    id: res.data.id,
                    real_name: res.data.real_name,
                    tokenValidateTime: Date.now(),
                }
                console.log(formatDate(new Date().getTime()) + ' [Authenticate] passed', client.id, client.connInfo)
            }
            done(null, true)
        } catch (err) {
            let msg = '[Authenticate Proceed - Err] '
            console.error(msg, err, client.busError)
            const error = new Error(msg)
            error.returnCode = 4
            done(error, null)
        }
        return
    }
}

/**
 * [主题]讯息订阅鉴权
 * @param {*} client 
 * @param {*} packet 
 * @param {*} done 
 * @returns 
 */
mqttServ.authorizeSubscribe = (client, subsInfo, done) => {
    if (client.busError instanceof ErrorResult) { //错误响应 - 修改为指定主题
        subsInfo.topic = 'ClientMsg/Client/' + client.id
        // subsInfo.qos = 1
        done(null, subsInfo)
        return
    }

    if (!!client.connInfo && client.connInfo.clientCat < 200) { // 客户端ACL
        // const topicInfo = topicCheck(subsInfo.topic, client)
        // if (topicInfo instanceof ErrorResult) {
        //     const msg = formatDate(new Date().getTime()) + ' [' + subsInfo.topic + '] 鉴权不通过, 原因:' + topicInfo.msg
        //     // console.log(msg)
        //     done(new Error(msg))
        //     return
        // }
    }

    // overwrites subscription
    // subsInfo.topic = 'foo'
    // subsInfo.qos = 2 //需求待开发

    done(null, subsInfo)
}

/**
 * [主题]讯息发布鉴权
 * @param {*} client 
 * @param {*} packet 
 * @param {*} done  
 * @returns 
 */
mqttServ.authorizePublish = (client, packet, done) => {

    if (!!client.connInfo && client.connInfo.clientCat < 200) { // 客户端ACL
        if (!packet.topic.startsWith(utils.$SYS_PREFIX) && packet.topic !== 'WillMsg') { //非系统操作的主题才需要过滤
            const res = topicCheck(packet.topic, client)
            if (res instanceof ErrorResult) {
                console.log('[authorizePublish]-packet.topic', packet.topic);
                done(new Error('[authorizePublish]-res.msg' + res.msg))
                return
            }
        }
    }
    //服务端发布不拦截

    //固定时间内, 类型有误次数 超过限制时, 中断连接     - dev...
    //不合法内容过滤                                  - dev...

    // overwrites publishing packet
    // packet.payload = Buffer.from('overwrite packet payload')

    done(null)
    // console.log('发布鉴权通过,published-payload:', packet.payload.toString())
}
console.log(formatDate(new Date().getTime()) + ' [node] mqttServ start')
